package com.kingstar.ceres.data.strategy;

import com.kingstar.ceres.data.bo.BaseFileMapping;
import com.kingstar.ceres.data.bo.CsvDataConfig;
import com.kingstar.ceres.data.constant.FileImportConstants;
import com.kingstar.ceres.data.handler.AbstractDataHandler;
import com.kingstar.ceres.data.handler.ITaskLockHandler;
import com.kingstar.ceres.data.properties.FileMappingProperties;
import com.kingstar.ceres.data.properties.FileParseProperties;
import com.kingstar.ceres.data.strategy.csv.BlockedCsvChannel;
import com.kingstar.ceres.data.strategy.csv.ICsvChannel;
import com.kingstar.ceres.data.strategy.csv.ICsvChannelManager;
import com.kingstar.ceres.data.util.FormatUtils;
import com.kingstar.ceres.data.util.MessageFormatUtils;
import com.kingstar.ceres.data.util.TextFileReader;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.StopWatch;

import java.io.IOException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
public class CsvChannelManager implements ICsvChannelManager {

    private List<ICsvChannel> channels=new ArrayList<>();
    private final AbstractDataHandler dataHandler;
    private final ITaskLockHandler taskLockHandler;
    private final FileParseProperties properties;

    public CsvChannelManager(AbstractDataHandler dataHandler, ITaskLockHandler taskLockHandler,FileParseProperties properties) {
        this.dataHandler = dataHandler;
        this.taskLockHandler = taskLockHandler;
        this.properties=properties;
    }

    @Override
    public void register(ICsvChannel csvChannel) {
        channels.add(csvChannel);
    }

    @Override
    public void process(BaseFileMapping baseFileMapping, String opDate) {
        FileMappingProperties.CsvFileMapping entity=(FileMappingProperties.CsvFileMapping) baseFileMapping;
        //1.匹配数据类型构建线程池
        List<InternalExecutor> list=new ArrayList<>();
        for (ICsvChannel channel : channels) {
            if (!channel.match(entity.getFileId())){
                continue;
            }
            list.add(new InternalExecutor(
                    new ThreadPoolExecutor(properties.getSingleCoreSize(), properties.getSingleMaxSize(),
                            properties.getSingleKeepAlive(), TimeUnit.SECONDS,
                            new LinkedBlockingDeque<>(properties.getSingleQueueSize()),
                            new ThreadPoolExecutor.CallerRunsPolicy()),
                    channel
            ));
        }
        //2.如果无任何匹配则跳过
        if (list.size()==0){
            log.warn("未匹配处理器:{}",entity);
            return;
        }
        StopWatch watch=new StopWatch();
        watch.start();
        //3.清空历史数据
        dataHandler.clearData(entity.getFileId(),opDate,entity.getClearStrategy());
        watch.split();
        //4.加载文件并提交线程池处理
        TextFileReader fileReader = null;
        try {
            Path path = Paths.get(entity.getLocalPath(), entity.getFileName());
            fileReader = new TextFileReader(path.toFile(),properties.getCsv().getCharset());
            if (!FileImportConstants.NO_HEAD.equals(entity.getHasHead())){
                log.info("文件有表头:{},{}",entity.getFileName(),entity.getHasHead());
                if (fileReader.hasNext()){
                    List<String> skips = fileReader.readNextLines(Integer.parseInt(entity.getHasHead()));
                    log.info("文件跳过表头行:{},data:{}",entity.getFileName(), skips);
                }
            }
            CsvDataConfig config = entity.toCsvDataConfig(opDate);
            while (fileReader.hasNext()){
                List<String> origins = fileReader.readNextLines(properties.getCsv().getReadLines());
                List<List<String>> payload = FormatUtils.parseCsv(origins, config.getRegex(), entity.getQuoteCharWrap());
                for (InternalExecutor internalExecutor : list) {
                    internalExecutor.executor.execute(()->{
                        internalExecutor.csvChannel.process(payload.stream().collect(Collectors.toList()), config);
                    });
                }
//                payload.clear();
            }

            shutdownAndBlock(entity, list);
            log.info("文件原始数据入库完成;文件名称:{},清除历史数据耗时:{}ms,入库耗时:{}ms",
                    entity.getFileName(),watch.getSplitTime(),watch.getTime());
            watch.stop();
            taskLockHandler.taskSuccess(entity.getFileType(),entity.getFileId(),opDate,entity.getCanRetry());
        } catch (IOException | InterruptedException e) {
            log.error(MessageFormatUtils.format("文件原始数据入库失败;文件名称:{}",
                    entity.getFileName()),e);
            taskLockHandler.fastFail(entity.getFileType(),entity.getFileId(),opDate,e.getMessage());
        }finally {
            if (fileReader!=null){
                try {
                    fileReader.close();
                } catch (IOException e) {
                    log.error("文件流关闭失败:",e);
                }
            }
        }
    }

    private void shutdownAndBlock(FileMappingProperties.CsvFileMapping entity, List<InternalExecutor> list) throws InterruptedException {
        ThreadPoolExecutor executor=null;
        for (InternalExecutor internalExecutor : list) {
            internalExecutor.executor.shutdown();
            if (internalExecutor.csvChannel instanceof BlockedCsvChannel){
                executor=internalExecutor.executor;
            }
        }
        if (executor!=null){
            while (!executor.isTerminated()){
                TimeUnit.SECONDS.sleep(properties.getSingleWaitInterval());
                log.info("数据入库任务执行中:{}", entity.getFileName());
            }
        }
    }

    @AllArgsConstructor
    private static class InternalExecutor{
        private ThreadPoolExecutor executor;
        private ICsvChannel csvChannel;
    }
}